package net.bwie.realtime.dy.dwd.log.funcation.zl.usercountfuncation;

import net.bwie.realtime.dy.dwd.log.bean.UserCountBean;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @ClassName：UserCountProcessorFuncation
 * @Author: ZddddQ
 * @Date: 2025/5/28 19:03
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
public class UserCountProcessorFuncation
        extends ProcessWindowFunction<Integer, UserCountBean, String, TimeWindow> {

    private final FastDateFormat fastDateFormat = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");

    @Override
    public void process(
            String roomId,
            Context context,
            Iterable<Integer> aggregates,
            Collector<UserCountBean> out) {

        // 从聚合结果中获取最终用户数
        Integer userCount = aggregates.iterator().next();

        // 获取窗口元数据
        TimeWindow window = context.window();
        String windowStart = fastDateFormat.format(window.getStart());
        String windowEnd = fastDateFormat.format(window.getEnd());

        // 输出带窗口信息的结果
        out.collect(new UserCountBean(
                windowStart,
                windowEnd,
                roomId,
                userCount,
                System.currentTimeMillis()
        ));
    }
}
